package com.zhike.utils;

import com.zhike.constant.KafkaConstant;
import com.zhike.function.MyKafkaSerialization;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.producer.ProducerConfig;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/18
 * @Description:
 */
public class KafkaUtils {
    //创建KafkaSource
    public static KafkaSource<String> getKafkaSource(String topic,String groupId) {
        return KafkaSource.<String>builder()
                .setBootstrapServers(KafkaConstant.KAFKA_BROKERS)
                .setTopics(topic)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
    }

    //创建KafkaSink
    public static KafkaSink<String> getKafkaSink(String topic){
        return KafkaSink.<String>builder()
                //指定Kafka的连接地址
                .setBootstrapServers(KafkaConstant.KAFKA_BROKERS)
                //指定序列化器
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic(topic)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                //写入kafka的一致性级别
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
//                //如果是精确一次，必须设置事务的前缀
//                .setTransactionalIdPrefix("zhike-")
//                //如果是精确一次必须设置事务超时时间
//                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "300000")
                .build();
    }
}

